#include "Producer.h"

#include <iostream>
#include "FileStream.h"

#include "rocksdb/file_system.h"


namespace streamdb {

class ProducerImp : public Producer{
public:
    ProducerImp(const std::string & dir, const std::string & topic) 
    : dir(dir)
    , topic(topic)
    , isOpen(false) 
    , stream(dir + "/" +  topic)
    {
        this->open();
    }
    ~ProducerImp()override{
        close();
    }

    Status open() override {
        // std::cout << stream.getName() << std::endl;
        //assert dir exist
        {
            auto fs = rocksdb::FileSystem::Default();
            auto s = fs->CreateDirIfMissing(dir, rocksdb::IOOptions(), nullptr);
            if(!s.ok()){
                std::cerr << "create dir failed" << s.ToString() << std::endl;
                return Status::Failed(s.ToString().c_str());
            }
        }
        //open stream 
        {
            auto s = stream.open();
            if(!s.isOk()){
                std::cerr << "stream open" << s.toString()<< std::endl;
                return s;
            }
        }

        isOpen = true;
        return Status::OK();
    }

    Status append(const std::string & object) override{
        if(isOpen){
            return stream.append(object);
        }
        std::cerr << "not opened" << std::endl;
        return Status::Failed("not opened");
    }
    void close() override {
        if(isOpen){
            stream.close();
            isOpen = false;
        }
    }

    const std::string & getDirectory() const override{
        return dir;
    }
    const std::string & getTopic() const override{
        return topic;
    }

private:
    std::string dir;
    std::string topic;
    bool isOpen;
    FileStream stream;
};


std::shared_ptr<Producer> makeProducer(const std::string & dir, const std::string & topic){
    return std::make_shared<ProducerImp>(dir, topic);
}


} //end of namespace streamdb
